package com.demo.flink

import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}
import org.apache.flink.streaming.api.windowing.time.Time

object WordCountStream {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //source
    val stream = env.socketTextStream("127.0.0.1", 9000, '\n')
    //transformation
    val counts = stream.flatMap(s=>s.split("\\s"))
      .map(word=>WordWithCount(word,1L))
      .keyBy("word")
      .timeWindow(Time.seconds(5), Time.seconds(1))
      .reduce((a,b)=>WordWithCount(a.word, a.count+b.count))
    //sink
    counts.print().setParallelism(1)
    env.execute("Socket Window WordCount")
  }

  case class WordWithCount(word: String, count: Long)
}
